// import org.apache.spark.mllib.linalg.{Matrices, Matrix}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.launcher.SparkLauncher
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
//注意,上面的Dataset是DataFrame的升级版,一般我们可以使用toDF来让Dataset转换回Dataframe
import org.apache.spark.sql.functions._ // for lit(), first(), etc.


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;



/**
  * Created by Administrator on 2017/4/20.
  * xudong
  */
object WordCountLocal {
  def main(args: Array[String]) {
    /**
      * SparkContext 的初始化需要一个SparkConf对象
      * SparkConf包含了Spark集群的配置的各种参数
      */

    Logger.getLogger("org").setLevel(Level.OFF)
    Logger.getLogger("akka").setLevel(Level.OFF)
    Logger.getRootLogger().setLevel(Level.ERROR)

    // val conf=new SparkConf().setAppName("wordcount").setMaster("spark://Desktop:7077");
    // //Spark程序的编写都是从SparkContext开始的
    // val sc=new SparkContext(conf)

    val spark = SparkSession.builder.appName("Intro").getOrCreate();
    spark.conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem")
    spark.conf.set("fs.file.impl","org.apache.hadoop.fs.LocalFileSystem")

    //以上的语句等价与val sc=new SparkContext("local","testRdd")
    val data=spark.sparkContext.textFile("hdfs://Desktop:9000/user/ds/hello.txt")//读取本地文件
    data.flatMap(_.split(" "))//下划线是占位符，flatMap是对行操作的方法，对读入的数据进行分割
      .map((_,1))//将每一项转换为key-value，数据是key，value是1
      .reduceByKey(_+_)//将具有相同key的项相加合并成一个
      .collect()//将分布式的RDD返回一个单机的scala array，在这个数组上运用scala的函数操作，并返回结果到驱动程序
      .foreach(println)//循环打印
  }
}

